package com.bw.ads;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.util.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

public class Test8 {
    public static void main(String[] args) throws Exception {
        //创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);

        DataStreamSource<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("tms_ods_yk6", "test691"));
        //1.当日发单总数，当日转运完成运单数，当日完成运输次数，当日完成运输里程，当日完成运输时长(10分)
        SingleOutputStreamOperator<Tuple6<String, Integer, Integer, Integer, Double, Long>> reduce = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {

                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    String table = jsonObject.getString("table");
                    if ("transport_task".equals(table) && jsonObject.getJSONObject("data") != null && jsonObject.getJSONObject("data").size() > 0) {
                        return true;
                    }
                } catch (Exception e) {
                }
                return false;
            }
        }).map(new MapFunction<String, Tuple6<String, Integer, Integer, Integer, Double, Long>>() {
            @Override
            public Tuple6<String, Integer, Integer, Integer, Double, Long> map(String s) throws Exception {
                JSONObject jo = JSON.parseObject(s).getJSONObject("data");
                String day = jo.getString("create_time").substring(0, 10);
                int order_cnt = 0; //单总数
                int complete_cnt = 0; //完成运单数
                int trans_cnt = 0; //完成运输次数
                double trans_dists = 0; //完成运输里程
                long trans_ts = 0; //完成运输时长

                order_cnt = jo.getInteger("order_num");
                if (jo.getString("status").equals("67004")) {
                    complete_cnt = jo.getInteger("order_num");
                    trans_cnt++;
                    trans_dists = jo.getDouble("actual_distance");
                    trans_ts = jo.getLong("actual_end_time") - jo.getLong("actual_start_time");
                }
                return new Tuple6<>(day, order_cnt, complete_cnt, trans_cnt, trans_dists, trans_ts);
            }
        }).keyBy(x -> x.f0).reduce((x, y) -> new Tuple6<>(x.f0, x.f1 + y.f1, x.f2 + y.f2, x.f3 + y.f3, x.f4 + y.f4, x.f5 + y.f5));
        reduce.print();

        env.execute();
    }
}
